package com.cneport.das.server.util.rmq;

import com.cneport.das.common.utils.JsonUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * @author GavinZhang
 * @version 1.0
 * @desc： RabbitMQ的使用帮助类
 * @company xxxxx 公司
 * @date 2021/8/17 15:29
 */
@Component
@Slf4j
public class RabbitMqHelper {

    @Autowired(required = false)
    private RabbitTemplate rabbitTemplate;
    /**
     * 配置确认回调
     */
    private final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        /**
         * @param correlationData 消息相关的数据，一般用于获取 唯一标识 id
         * @param b true 消息确认成功，false 失败
         * @param s 确认失败的原因
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean b, String s) {
            if (b) {
                log.info("confirm 消息确认成功..." + correlationData.getId());
            } else {
                log.info("confirm 消息确认失败..." + correlationData.getId() + " cause: " + s);
            }
        }
    };

    /**
     * 配置Return机制 的回调方法（找不到路由才会触发）
     */
    private final RabbitTemplate.ReturnsCallback returnCallback = new RabbitTemplate.ReturnsCallback() {
        @Override
        public void returnedMessage(ReturnedMessage returnedMessage) {
            StringBuilder sb = new StringBuilder("RabbitMQ触发了Return机制\n");
            sb.append("ReplyText:" + returnedMessage.getReplyText() + "\n");
            sb.append("Exchange:" + returnedMessage.getExchange() + "\n");
            sb.append("RoutingKey:" + returnedMessage.getRoutingKey() + "\n");
            sb.append("Message:" + JsonUtil.toJsonString(returnedMessage.getMessage()) + "\n");
            log.info(sb.toString());
        }
    };


    /**
     * 发送消息
     *
     * @param rabbitTemplate
     * @param queueName
     * @param message
     */
    public void sendMsg(RabbitTemplate rabbitTemplate, String queueName, String message) {
        rabbitTemplate.convertAndSend(queueName, message);
    }

    /**
     * 发送消息
     *
     * @param queueName
     * @param message
     */
    public void sendMsg(String queueName, String message) {
        rabbitTemplate.convertAndSend(queueName, message);
    }

    /**
     * 发送消息
     *
     * @param rabbitTemplate
     * @param exchange
     * @param routingKey
     * @param message
     */
    public void sendMsg(RabbitTemplate rabbitTemplate, String exchange, String routingKey, String message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
    }

    /**
     * 发送消息
     *
     * @param exchange
     * @param routingKey
     * @param message
     */
    public void sendMsg(String exchange, String routingKey, String message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
    }

    /**
     * 带确认的发送
     *
     * @param rabbitTemplate
     * @param queueName
     * @param msgObj
     */
    public void sendWithConfirm(RabbitTemplate rabbitTemplate, String queueName, Object msgObj) {
        rabbitTemplate.convertAndSend(queueName, msgObj, new CorrelationData("" + System.currentTimeMillis()));
        rabbitTemplate.setConfirmCallback(confirmCallback);
    }

    /**
     * 带确认的发送
     *
     * @param queueName
     * @param msgObj
     */
    public void sendWithConfirm(String queueName, Object msgObj) {
        rabbitTemplate.convertAndSend(queueName, msgObj, new CorrelationData("" + System.currentTimeMillis()));
        rabbitTemplate.setConfirmCallback(confirmCallback);
    }

    /**
     * 带Return的发送，找不到路由时，路由到这里
     *
     * @param exchange
     * @param routingKey
     * @param msgObj
     */
    public void sendWithReturn(String exchange, String routingKey, Object msgObj) {
        rabbitTemplate.setReturnsCallback(returnCallback);
        rabbitTemplate.convertAndSend(exchange, routingKey, msgObj);
    }

    /**
     * 带Return的发送，找不到路由时，路由到这里
     *
     * @param rabbitTemplate
     * @param exchange
     * @param routingKey
     * @param msgObj
     */
    public void sendWithReturn(RabbitTemplate rabbitTemplate, String exchange, String routingKey, Object msgObj) {
        rabbitTemplate.setReturnsCallback(returnCallback);
        rabbitTemplate.convertAndSend(exchange, routingKey, msgObj);
    }


    /**
     * 收取消息
     *
     * @param queueName
     * @param cnt       一次取cnt数据，若不够cnt，就取到所有的。
     * @return
     */
    public String receive(String queueName, int cnt) {
        return receive(rabbitTemplate, queueName, cnt);
    }

    /**
     * 收取消息
     *
     * @param rabbitTemplate
     * @param queueName
     * @param cnt            一次取cnt数据，若不够cnt，就取到所有的。
     * @return
     */
    public String receive(RabbitTemplate rabbitTemplate, String queueName, int cnt) {
        if (cnt == 0 || cnt > 10) {
            cnt = 1;
        }

        ConnectionFactory connectionFactory = rabbitTemplate.getConnectionFactory();
        Connection connection = connectionFactory.createConnection();
        Channel channel = connection.createChannel(false);
        List<MessageEntity> list = new ArrayList<>();
        long start = System.currentTimeMillis();
        long now = System.currentTimeMillis();
        List<String> rtn = new ArrayList();
        while (true) {
            try {
                GetResponse message = channel.basicGet(queueName, false);

                long nowTime = System.currentTimeMillis();
                long interval = nowTime - start;

                if (message != null) {
                    MessageEntity entity = new MessageEntity();
                    String text = new String(message.getBody(), StandardCharsets.UTF_8);
                    long tag = message.getEnvelope().getDeliveryTag();
                    entity.setMessage(text);
                    entity.setTag(tag);
                    list.add(entity);
                }

                if (list.size() == cnt || interval > 200) {
                    //批量确认
                    for (MessageEntity en : list) {
                        channel.basicAck(en.getTag(), true);
                        rtn.add(en.getMessage());
                    }
                    list.clear();
                    return JsonUtil.toJsonString(rtn);
                }

            } catch (Exception e) {
                e.printStackTrace();
                break;
            } finally {
                connection.close();
            }


        }
        return "";
    }
}
